/*
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or (at
your option) any later version.

This program is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
02110-1301, USA.
*/
// Copyright 2010 Tencent Inc.
// Author: Yi Wang (yiwang@tencet.com)
//
#ifndef SORTED_BUFFER_SORTED_BUFFER_ITERATOR_H_
#define SORTED_BUFFER_SORTED_BUFFER_ITERATOR_H_

#include <string>
#include <vector>

#include "src/base/common.h"
#include "src/sorted_buffer/memory_piece.h"

namespace sorted_buffer {

// The interface of iterator.
class SortedBufferIterator {
 public:
  virtual ~SortedBufferIterator() {}
  virtual const std::string& key() const = 0;
  virtual const std::string& value() const = 0;
  virtual bool Done() const = 0;          // Done with values of current key.
  virtual void Next() = 0;                // Jump to the next value
  virtual void DiscardRestValues() = 0;   // Jump until all values are skipped.
};


// Traverse disk files generated by SortedBuffer for sorted map outputs.
class SortedBufferIteratorImpl : public SortedBufferIterator {
 public:
  SortedBufferIteratorImpl(const std::string& filebase, int num_files);
  virtual ~SortedBufferIteratorImpl();

  virtual const std::string& key() const;
  virtual const std::string& value() const;
  virtual bool Done() const;          // Done with values of current key.
  virtual void Next();                // Jump to the next value of current key
  virtual void DiscardRestValues();   // Jump skip all values of current key.

  void NextKey();              // Jump to the next reduce input (key).
  bool FinishedAll() const;    // Done with all keys and values.

 private:
  struct SortedStringFile {
    FILE* input;
    int index;
    std::string top_key;
    std::string top_value;
    int32 num_rest_values;  // number of values of top_key left in current
                            // file. 0 means no value for the key on disk
                            // but might be one in top_key.  Negative
                            // value means "end-of-sorted_buffer".
  };

  struct SSFileCompare {
    bool operator() (const SortedStringFile* lhs,
                     const SortedStringFile* rhs) {
      return lhs->top_key > rhs->top_key;
    }
  };

  typedef std::vector<SortedStringFile*> SSFileList;

  std::string current_key_;
  std::string filebase_;
  SSFileCompare compare_;
  SSFileList files_;  // A min-heap structure is maintained inside
  int heap_size_;
  bool done_;

  // Invoked by ctor. Open all block files (specified by filebase and
  // num_files).  Requires that each file contains at least one key-value pair.
  void Initialize(const std::string& filebase, int num_files);

  // Invoked by dtor.
  void Clear();

  // Returns false if file->num_rest_values <= 0
  bool LoadValue(SortedStringFile* file);

  // If no more keys exist in the file, mark it as end-of-sorted_buffer
  // and returns false.
  bool LoadKey(SortedStringFile* file);

  // Swap file pointer in files_ list
  void SwapFile(int i, int j);

  // A min-heap structure is maintained in files_ list, where files_[0] is the
  // top element. The heap structure is broken after the key of top file is
  // updated, sift down the top element so that the heap structure is still kept
  void SiftDown();

};

}  // namespace sorted_buffer

#endif  // SORTED_BUFFER_SORTED_BUFFER_ITERATOR_H_
